package com.giw.project.simple_graph_traversal;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text; //import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.giw.project.exception_manager.UncheckedException;


public class TraversalApp extends Configured implements Tool {

	private final static String tmp_path = "tmp.txt";
	private final static String tmp_path2 = "tmp2.txt";
	private final static String tmp_pathS = "tmp_searchKind.txt";
	private final static String absolute_path="/home/giorgio/Scrivania/GIW/progetto/";
	private final static Map<String,String> node2input=new HashMap<String,String>();
	private String uriOriginalMovie;



	private void initializeMap(){
		node2input.put("string",absolute_path+"litteral");
		node2input.put("<http://data.linkedmdb.org/resource/film/", absolute_path+"movie");
		node2input.put("<http://data.linkedmdb.org/resource/director/", absolute_path+"director");
		node2input.put("<http://data.linkedmdb.org/resource/film_genre/", absolute_path+"genre");
		node2input.put("<http://data.linkedmdb.org/resource/actor/", absolute_path+"actor");
		node2input.put("else", absolute_path + "else3");

	}
	public int run(String[] args) throws Exception {
		this.initializeMap();
		boolean success = true;
		int count_iteration = 0;

		File fS = new File(tmp_pathS);
		FileOutputStream fosS = new FileOutputStream(fS);
		PrintStream outputS = new PrintStream(fosS);
		outputS.println(args[2]);
		outputS.close();
		fosS.close();

		File f = new File(tmp_path);
		FileOutputStream fos = new FileOutputStream(f);
		PrintStream output = new PrintStream(fos);
		output.println(args[1].toLowerCase() + "\t" + "<http://purl.org/dc/terms/title>");
		output.close();
		fos.close();
		String input;

		while (f.exists()) {
			count_iteration ++;
			if(count_iteration==1){
				input=node2input.get("string");
			}
			else{
				input=this.getInputDirectory();
				if(input==null){
					throw new UncheckedException("There's no an input for the that kind of node," +
					"see the tmp.txt file");
				}
			}
			if(count_iteration==2){
				this.getUriOriginalMovie();
			}
			Job job = new Job(getConf());
			job.setJobName("Graph_Traversal");
			job.setJarByClass(TraversalMapper.class);
			job.setMapperClass(TraversalMapper.class);
			job.setReducerClass(TraversalReducer.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(Text.class);
			job.setInputFormatClass(TextInputFormat.class);
			job.setOutputFormatClass(TextOutputFormat.class);
			FileInputFormat.addInputPath(job, new Path(input));
			FileInputFormat.addInputPath(job, new Path(absolute_path + "movie"));
			FileOutputFormat.setOutputPath(job, new Path(args[0]+count_iteration));

			success = job.waitForCompletion(true);
			f.delete();
			File file1 = new File(tmp_path2);
			File file2 = new File(tmp_path);
			file1.renameTo(file2);

		}
		int i=1;
		for(i=1;i<count_iteration;i++){
			f=new File(args[0]+count_iteration);
			f.delete();
		}
		f = new File(tmp_path2);
		f.delete();
		f = new File(tmp_pathS);
		f.delete();
		this.eliminateOriginalMovie(args[0]+count_iteration);
		return success ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {
		if (args.length != 3) {
			System.err
			.println("Usage: <output path> <movie-title> <search-kind>");
			// search kind = genre or director or actors
			System.exit(-1);
		}
		int ret = ToolRunner.run(new TraversalApp(), args);
		System.exit(ret);
	}

	public String getInputDirectory(){
		File f=new File(tmp_path);
		FileInputStream fis;
		String line=null;
		try {
			fis = new FileInputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException("No input directory!\nMessage:"+e.getMessage());
		}
		InputStreamReader is=new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		try {
			line=buffer.readLine();
		} catch (IOException e) {
			throw new UncheckedException("No line to read!\nMessage:"+e.getMessage());
		}
		try {
			buffer.close();
			is.close();
			fis.close();
		} catch (IOException e) {
			e.printStackTrace();
		}

		String inputDirectory=node2input.get(this.getNodeKind(line.split("\t")[0]));
		if(inputDirectory==null){
			inputDirectory=node2input.get("else");
		}
		return inputDirectory;
	}


	private String getNodeKind(String uri) {
		if (!uri.contains("http")) {
			return "string";
		}
		String[] parts_uri = uri.split("/");
		int i;
		uri = "";
		for (i = 0; i < parts_uri.length - 1; i++) {
			uri += "/" + parts_uri[i];
		}
		return uri.substring(1) + "/";

	}

	private void getUriOriginalMovie(){
		File f=new File(tmp_path);
		FileInputStream fis;
		String line=null;
		try {
			fis = new FileInputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException(e.getMessage());
		}
		InputStreamReader is=new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		try {
			line=buffer.readLine();
		} catch (IOException e) {
			throw new UncheckedException(e.getMessage());
		}
		try {
			buffer.close();
			is.close();
			fis.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		this.uriOriginalMovie=line.split("\t")[0];	
	}

	private void eliminateOriginalMovie(String path){
		List<String> results=new LinkedList<String>();
		File f=new File(path+"/part-r-00000");
		FileInputStream fis;
		String line=null;
		try {
			fis = new FileInputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException(e.getMessage());
		}
		InputStreamReader is=new InputStreamReader(fis);
		BufferedReader buffer = new BufferedReader(is);
		try {
			line=buffer.readLine();
			while(line!=null){
				if(!(line.split("\t")[0].equals(this.uriOriginalMovie))){
					results.add(line);
				}
				line=buffer.readLine();
			}
		} catch (IOException e) {
			throw new UncheckedException(e.getMessage());
		}
		try {
			buffer.close();
			is.close();
			fis.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		
		f.delete();
		f = new File(path+"/part-r-00000");
		FileOutputStream fos;
		try {
			fos = new FileOutputStream(f);
		} catch (FileNotFoundException e) {
			throw new UncheckedException(e.getMessage());
		}
		PrintStream output = new PrintStream(fos);
		for(String result:results){
		output.println(result);
		}
		output.close();
		try {
			fos.close();
		} catch (IOException e) {
			throw new UncheckedException(e.getMessage());
		}
	}

}
